package beautiful.butterfly.drds.binlog.connection;

import beautiful.butterfly.drds.binlog.Zookeeper;
import beautiful.butterfly.drds.binlog.event.data.TableMapEventData;
import beautiful.butterfly.drds.binlog.snapshot.SnapShot;
import beautiful.butterfly.drds.binlog.snapshot.SnapShotType;
import beautiful.butterfly.drds.binlog.type.CheckSumType;
import beautiful.butterfly.drds.binlog.utils.*;
import beautiful.butterfly.drds.binlog.zookeeper.ZooKeepers;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;

@Slf4j
public class ConnectionAttributes
{


    @Setter
    @Getter
    private boolean firstFormatDescription = true;
    @Setter
    @Getter
    private int nettyManageState = 0;// 0：放入体系 1:成功 -1:失败
    @Zookeeper
    @Setter
    @Getter
    private String runningPath; //// runningPath
    @Setter
    @Getter
    private String binlogPositionPath; // binlogPositionPath
    @Setter
    @Getter
    private long clientId;// clientId
    @Setter
    @Getter
    private String ip; // 主机
    @Setter
    @Getter
    private int port;
    @Setter
    @Getter
    private String realConnectedIP;// 多个IP前提下，真正连接的有效IP
    @Setter
    @Getter
    private String usernameForReplication; // 用于复制
    @Setter
    @Getter
    private String passwordForReplication;
    @Setter
    @Getter
    private String usernameForSchema; // 用于获取schemaS
    @Setter
    @Getter
    private String passwordForSchema;
    @Setter
    @Getter
    private CheckSumType checkSumType = CheckSumType.none; // checksum
    // 过滤专用
    @Setter
    @Getter
    private List<DatabaseNameAndTableNameFilter> databaseNameAndTableNameFilterList; // filter rules
    @Setter
    @Getter
    private HashMap<String, Boolean> learnedFilterRules = new HashMap<String, Boolean>();
    //
    // 分库分表专用
    @Setter
    @Getter
    private List<ShardingRule> shardingRuleList;// shardingRuleList
    @Setter
    @Getter
    private HashMap<String, ShardingRule> learnedShardingRules = new HashMap<String, ShardingRule>();
    //
    // 用来保存上一次处理的位置
    @Setter
    @Getter
    private String binlogFileName = null; // binlog
    @Setter
    @Getter
    private long binlogFilePosition = 4;
    //
    // 用来处理ZK History的逻辑
    @Setter
    @Getter
    private Long currentPosition;// 本条数据的position
    @Setter
    @Getter
    private Long nextPosition;// 上条数据的的position
    @Setter
    @Getter
    private String historyPositionDay;// 上次存放position的日期: 20160126
    //
    //
    //
    // 全局有效标志,涉及到多个线程的交互，所以一定满足互斥
    // 默认为true,一旦为false,所有相关方应该立刻停止处理
    @Setter
    @Getter
    private AtomicBoolean globalValid = new AtomicBoolean(true);
    //
    //
    // 3)tableIdToTableMapEventDataMap
    // 临时性的事件处理,用完了应该立刻删除,防止内存占用过多
    @Setter
    @Getter
    private HashMap<Long, TableMapEventData> tableIdToTableMapEventDataMap = new HashMap<Long, TableMapEventData>();
    // 4)保留本连接对应的数据库-表-列名的关系// 处理database-tableName-columns映射关系
    @Setter
    @Getter
    private HashMap<String, HashMap<String, String>> databaseNameAndTableNameToPositionToColumnNameMapMap = new HashMap<String, HashMap<String, String>>();
    // 记载了每个表的主键关系,如果没有，则应该为空
    @Setter
    @Getter
    private HashMap<String, ArrayList<String>> databaseNameAndTableNameToPrimaryKeyColumnNameListMap = new HashMap<String, ArrayList<String>>();
    // 处理各种snapshot信息
    @Setter
    @Getter
    private SnapShot taskHeader = null;
    @Setter
    @Getter
    private SnapShot taskTailer = null;
    @Setter
    @Getter
    private Thread taskThread;

    public ConnectionAttributes(SnapShot snapShot)
    {
        // 刚开始初始化为一样的
        taskHeader = snapShot;
        taskTailer = snapShot;
    }

    public static ConnectionAttributes parse(String data)
    {
        ConnectionAttributes connectionAttributes = null;
        // 提取出各种k/v
        JSONObject jsonObject = JSON.parseObject(data);
        String ur = jsonObject.getString(Constants.USERNAME_FOR_REPLICATION);
        String pr = jsonObject.getString(Constants.PASSWORD_FOR_REPLICATION);
        String us = jsonObject.getString(Constants.USERNAME_FOR_SCHEMA);
        String ps = jsonObject.getString(Constants.PASSWORD_FOR_SCHEMA);
        // 开始构建库表过滤规则
        ArrayList<DatabaseNameAndTableNameFilter> tempDatabaseNameAndTableNameFilters = new ArrayList<DatabaseNameAndTableNameFilter>();
        {
            JSONArray rulesArray = jsonObject.getJSONArray(Constants.FILTER_RULES);
            JSONObject rule;
            String database;
            String table;
            for (Object obj : rulesArray)
            {
                rule = (JSONObject) obj;
                database = rule.getString(Constants.DATABASE_FOR_FILTER_RULES);
                table = rule.getString(Constants.TABLE_FOR_FILTER_RULES);
                tempDatabaseNameAndTableNameFilters.add(new DatabaseNameAndTableNameFilter(Pattern.compile(database), Pattern.compile(table)));
            }
        }
        // 开始构建分库分表规则
        ArrayList<ShardingRule> tempShardingRules = new ArrayList<ShardingRule>();
        {
            JSONArray rulesArray = jsonObject.getJSONArray(Constants.SHARDING_RULES);
            JSONObject rule;
            String regexDatabase;
            String regexTable;
            String targetDatabase;
            String targetTable;
            for (Object obj : rulesArray)
            {
                rule = (JSONObject) obj;
                regexDatabase = rule.getString(Constants.REGEX_DATABASE_FOR_SHARDING_RULES);
                regexTable = rule.getString(Constants.REGEX_TABLE_FOR_SHARDING_RULES);
                targetDatabase = rule.getString(Constants.TARGET_DATABASE_FOR_SHARDING_RULES);
                targetTable = rule.getString(Constants.TARGET_TABLE_FOR_SHARDING_RULES);
                tempShardingRules.add(new ShardingRule(//
                        Pattern.compile(regexDatabase), //
                        Pattern.compile(regexTable), //
                        targetDatabase, //
                        targetTable));//
            }
        }
        // 开始构造对象并赋值
        connectionAttributes = new ConnectionAttributes(
                new SnapShot(null, SnapShotType.none, false, null, null, null, null, null, null));
        connectionAttributes.setUsernameForReplication(ur);
        connectionAttributes.setPasswordForReplication(pr);
        connectionAttributes.setUsernameForSchema(us);
        connectionAttributes.setPasswordForSchema(ps);
        // 设置filter规则
        connectionAttributes.setDatabaseNameAndTableNameFilterList(tempDatabaseNameAndTableNameFilters);
        // 设置sharding规则
        connectionAttributes.setShardingRuleList(tempShardingRules);
        // 返回对象
        return connectionAttributes;

    }


    public void setGlobalValid(Boolean globalValid)
    {
        this.globalValid.set(globalValid);
    }


    // RunningPath
    public ConnectionAttributes setRunningPath(String runningPath)
    {
        this.runningPath = runningPath;
        return this;
    }


    // BinlogPositionZKPath
    public ConnectionAttributes setBinlogPositionPath(String binlogPositionPath)
    {
        this.binlogPositionPath = binlogPositionPath;
        return this;
    }


    // ClientId
    public ConnectionAttributes setClientId(long clientId)
    {
        this.clientId = clientId;
        log.debug("clientId:" + clientId);
        return this;
    }

    // IP&Port
    public ConnectionAttributes setIpPort(String ip, int port, String rcIP)
    {
        this.ip = ip;
        this.port = port;
        this.realConnectedIP = rcIP;
        return this;
    }


    public long getChecksumLength()
    {
        return this.checkSumType.getLength();
    }


    public boolean acceptByFilter(String databaseName, String tableName)
    {
        // 先看本地是否学习过
        String databaseNameAndTableName = StringUtils.union(databaseName, tableName);
        Boolean learnedResult = learnedFilterRules.get(databaseNameAndTableName);
        if (null != learnedResult)
        {

            return learnedResult.booleanValue();
        }

        // 之前没有学习过，遍历本地rules
        boolean result = false;
        if (databaseNameAndTableNameFilterList.isEmpty())
        {
            result = true;
        } else
        {
            for (DatabaseNameAndTableNameFilter databaseNameAndTableNameFilter : databaseNameAndTableNameFilterList)
            {
                if (databaseNameAndTableNameFilter.accept(databaseName, tableName))
                {
                    result = true;
                    break;
                }
            }
        }
        // 返回前，存入学习结果[无论成功还是失败,都是学习结果]
        learnedFilterRules.put(databaseNameAndTableName, result);
        return result;

    }

    public ShardingRule findShardingRule(String d, String t)
    {
        // 先看本地是否学习过
        String key = StringUtils.union(d, t);
        // 之前已经存在过结果，可能是一个规则，也可能是null[表示没有对应的分库分表规则]
        if (learnedShardingRules.containsKey(key))
        {
            return learnedShardingRules.get(key);
        }
        // 之前没有学习过，遍历本地rules
        // LoggerUtils.debug(logger, "no learned regex,learn from origin rules:"
        // + rules);
        //
        ShardingRule result = null;
        for (ShardingRule shardingRule : shardingRuleList)
        {
            if (shardingRule.accept(d, t))
            {
                result = shardingRule;
                break;
            }
        }
        // 返回前，存入学习结果[无论成功还是null,都是学习结果]
        learnedShardingRules.put(key, result);
        //
        return result;
    }

    // binlogFileName
    public ConnectionAttributes updateBinlogFileNameAndPosition(String name, long position)
    {
        this.binlogFileName = name.trim();
        this.binlogFilePosition = position < 4 ? 4 : position;
        return this;
    }


    public long getBinlogFilePosition()
    {
        if (binlogFilePosition < 4)
            binlogFilePosition = 4;
        return binlogFilePosition;
    }

    public void putTableMapEventData(long tableId, TableMapEventData tableMapEventData)
    {
        log.debug("tableId:" + tableId + " " + tableMapEventData.getDatabaseName() + " " + tableMapEventData.getTableName() + tableMapEventData.getTableId());
        tableIdToTableMapEventDataMap.put(tableId, tableMapEventData);
    }

    public TableMapEventData getTableMapEventData(long tableId)
    {
        return tableIdToTableMapEventDataMap.get(tableId);
    }

    public void deleteTableMapEventData()
    {
        // 直接删除了
        tableIdToTableMapEventDataMap.clear();
    }


    public ArrayList<String> getPrimaryKeyColumnNameList(String databaseName, String tableName)
    {
        String databaseNameAndTableName = StringUtils.union(databaseName, tableName);
        return databaseNameAndTableNameToPrimaryKeyColumnNameListMap.get(databaseNameAndTableName);
    }

    public void ensureDatabaseTableColumnsMappingDeleted(String databaseName, String tableName)
    {
        String databaseNameAndTableName = StringUtils.union(databaseName, tableName);
        this.databaseNameAndTableNameToPositionToColumnNameMapMap.remove(databaseNameAndTableName);
        this.databaseNameAndTableNameToPrimaryKeyColumnNameListMap.remove(databaseNameAndTableName);
    }

    public void ensureDatabaseTableColumnsMappingExist(String databaseName, String tableName, boolean forceUpdate)
            throws Exception
    {
        // LoggerUtils.info(logger, "[" + database + "][" + tableName + "]
        // forceUpdate:" + forceUpdate);
        // 因为比较耗时，所幸,并不是经常改数据表结构
        // 如果之前拉过一次，大部分情况后面不需要再重新拉取
        String databaseNameAndTableName = StringUtils.union(databaseName, tableName);
        if (false == forceUpdate)
        {// 不强制更新，有就行了
            if (null != databaseNameAndTableNameToPositionToColumnNameMapMap.get(databaseNameAndTableName))
            {
                // 已经有了，不用做其它操作
                // LoggerUtils.debug(logger, "mappings已经存在，不用更新");
            } else
            {
                HashMap<String, String> positionToColumnNameMap = SqlUtils.getDatabaseTableColumnsMapping(//
                        realConnectedIP,//
                        port,//
                        this.usernameForSchema,//
                        this.passwordForSchema, //
                        databaseName,//
                        tableName, //
                        databaseNameAndTableNameToPrimaryKeyColumnNameListMap);//
                // LoggerUtils.debug(logger, "非强制更新" + mappings);
                databaseNameAndTableNameToPositionToColumnNameMapMap.put(databaseNameAndTableName, positionToColumnNameMap);
            }
        } else
        {
            // 强制更新,拉取强制更新,不管有没有，一律强制更新
            // 比如修改了表结构[这种情况也不多,没事修改表结构干嘛。。。:)]
            HashMap<String, String> positionToColumnNameMap = SqlUtils.getDatabaseTableColumnsMapping(//
                    realConnectedIP,//
                    port,//
                    usernameForSchema,//
                    passwordForSchema, //
                    databaseName, //
                    tableName, //
                    databaseNameAndTableNameToPrimaryKeyColumnNameListMap);//
            // LoggerUtils.debug(logger, "强制更新" + mappings);
            // put会替换原有的值
            databaseNameAndTableNameToPositionToColumnNameMapMap.put(databaseNameAndTableName, positionToColumnNameMap);
        }
    }


    public HashMap<String, String> getPositionToColumnNameMap(String databaseName, String tableName)
    {
        return databaseNameAndTableNameToPositionToColumnNameMapMap.get(StringUtils.union(databaseName, tableName));
    }

    public void addSnapShotToTail(SnapShot snapShot)
    {
        // 增加对全局唯一变量的有效性的检测
        if (false == snapShot.getGlobalValid().get())
        {
            return;
        }
        taskTailer.setNextSnapShot(snapShot);
        taskTailer = snapShot;
    }


    public void startTaskThread()
    {
        Runnable runnable = new Runnable()
        {// 只属于观察者模式,不修改任何变量的值.ReadOnly
            // binlogzkpath从这个里面取
            String currentValues = null;
            int accumulatedIOEvent = 0;
            long begin = System.currentTimeMillis();

            @Override
            public void run()
            {
                // 条件性的无限循环中
                log.debug("task Thread started...");
                //
                while (true == globalValid.get())
                {
                    SnapShot snapShot = taskHeader.getNextSnapShot();
                    if (null == snapShot)
                    {
                        TimeUtils.sleepMilliSeconds(10);// 主动休眠10ms
                        continue;
                    }
                    // 拿到了一个有效的snapshot
                    SnapShotType snapShotType = snapShot.getSnapShotType();
                    if (SnapShotType.rotate == snapShotType)
                    {
                        // 无条件更新
                        log.info("Rotate event");
                        ZooKeepers.update(binlogPositionPath, snapShot.getZkValue());
                        currentValues = null;
                        accumulatedIOEvent = 0;
                        begin = System.currentTimeMillis();
                    } else if (SnapShotType.io == snapShotType)
                    {// IO类型

                        while (true == globalValid.get()
                                && WriteResultUtils.PENDING.ordinal() == snapShot.getWriteResult())
                        {
                            // do nothing
                            // 一直等待写入结果
                            // 一切安好，否则退出
                            TimeUtils.sleepMilliSeconds(1);
                        }
                        if (true == globalValid.get() && WriteResultUtils.SUCCEED.ordinal() == snapShot.getWriteResult())
                        {
                            // 确实写入成功
                        } else
                        {// 有情况,线程退出
                            break;
                        }

                        // 考虑到按行加速,有可能是null
                        // 只负责更新currentValues
                        accumulatedIOEvent += (snapShot.getDatas().size() > 0 ? snapShot.getDatas().size() : 1);
                        // LoggerUtils.info(logger,
                        // "increase " + shot.getDatas().size() + " message,to
                        // be " + accumulatedCount);
                        if (null != snapShot.getZkValue())
                        {
                            currentValues = snapShot.getZkValue();
                        }

                    } else if (SnapShotType.xid == snapShotType)
                    {// xid类型
                        // 只负责在超过阀值时，更新ZK,保证此位置肯定是一个XID的位置
                        // LoggerUtils.info(logger, "xid event");
                        if (null != currentValues
                                && accumulatedIOEvent >= PropertiesExtend.getInstance().getAccumalatedCountValue())
                        {
                            // 更新ZK
                            if (begin > 0)
                                log.info("update  binlog current position , cost xxx: "
                                        + (System.currentTimeMillis() - begin) + " ms, count:" + accumulatedIOEvent
                                        + " --- " + binlogPositionPath + ":" + currentValues);
                            ZooKeepers.update(binlogPositionPath, currentValues);
                            currentValues = null;
                            accumulatedIOEvent = 0;
                            begin = System.currentTimeMillis();
                        }
                    } else
                    {
                        // what is wrong?
                        log.info("unknown event snapShotType in ConnectionAttributes thread...");
                    }
                    // 更新taskHeader的位置
                    taskHeader = snapShot;
                }
                //
                // 释放链接关系,促进垃圾回收
                taskHeader.setNextSnapShot(null);
                log.info("task Thread exit...");
            }
        };
        taskThread = new Thread(runnable);
        taskThread.start();
    }

}